package com.avo.ramcache.persist.disruptor;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.persist.Persister;
import com.avo.ramcache.persist.disruptor.event.ElementEvent;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.dsl.Disruptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

public class DisruptorConsumer implements EventHandler<ElementEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorConsumer.class);
    private final Persister owner;
    private final Accessor accessor;
    private final String name;
    private final Disruptor<ElementEvent> disruptor;
    private final ExecutorService executor;
    private volatile boolean stoped;
    private final AtomicInteger error = new AtomicInteger();

    public DisruptorConsumer(String name, Accessor accessor, Disruptor<ElementEvent> disruptor, ExecutorService executor, Persister owner) {
        this.owner = owner;
        this.accessor = accessor;
        this.disruptor = disruptor;
        this.executor =executor;
        this.name = name;
    }

    @Override
    public void onEvent(ElementEvent event, long sequence, boolean endOfBatch) throws Exception {
        if (stoped) {
            return;
        }
        Element element = event.getElement();
        if (element == null) {
            return;
        }
        persist(element);
    }

    private void persist(Element element) {
        IEntity<?> entity = element.getEntity();
        Class clz = element.getEntityClass();
        Listener listener;
        boolean succFlag = false;
        try {
            switch (element.getType()) {
                case SAVE:
                    if (entity.serialize()) {
                        accessor.saveOrUpdate(clz, entity);
                        succFlag = true;
                    }
                    break;
                case UPDATE:
                    if (entity.serialize()) {
                        accessor.saveOrUpdate(clz, entity);
                        succFlag = true;
                    }
                    break;
                case REMOVE:
                    accessor.remove(clz, entity.getId());
                    succFlag = true;
            }

            listener = this.owner.getListener(clz);
            if (listener != null) {
                listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), null);
            }
        } catch (RuntimeException e) {
            e.printStackTrace();
            this.error.getAndIncrement();
            LOGGER.error("实体更新队列{" + name + "}处理元素{" + element + "}时data{" + entity + "}出现异常", e);
            listener = this.owner.getListener(clz);
            if (listener != null) {
                listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), e);
            }
        } catch (Exception e) {
            this.error.getAndIncrement();
            LOGGER.error("更新队列出现未知异常", e);
        }
    }

    public void stop() {
        stoped = true;
        executor.shutdown();
        disruptor.shutdown();
        // try {
        //      disruptor.shutdown();
        //     disruptor.shutdown(10, TimeUnit.SECONDS);
        // } catch (TimeoutException e) {
        //     LOGGER.error("disruptor shutdown error", e);
        // }
    }
}
